package com.g.ocache.impl.store;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;

import com.g.ocache.Cache;
import com.g.ocache.CacheValidator;
import com.g.ocache.StoreEntry;
import com.g.ocache.ValidationException;
import com.g.ocache.event.StoreEventListener;
import com.g.ocache.event.StoreEventType;
import com.g.ocache.impl.store.expiration.Expirable;
import com.g.ocache.impl.store.expiration.ExpirationEvictor;


/**
 * 
 * @author zhangyf
 * @param <K>
 * @param <V>
 */
public abstract class AbstractSimpleStore<K, V> extends AbstractStore<K, V> implements Expirable<K, V> {
	//
	protected static final int DEFAULT_INITIAL_CAPACITY = 128;
	protected static final float DEFAULT_GROWTH_FACTOR = 0.75F;
	
	//
	private final int capacity;
	private final AtomicBoolean lenient;
	private final ReentrantReadWriteLock lock;
	private final ExpirationEvictor<K, V> evictor;
	private final StoreEntry.Factory<K, V> entryFactory;
	
	//
	protected abstract int doSize();
	protected abstract void doClear();
	protected abstract StoreEntry<K, V> doRemove(K key);
	protected abstract StoreEntry<K, V> doGet(K key);
	protected abstract Map<K, StoreEntry<K, V>> doGetAll();
	protected abstract StoreEntry<K, V> doEvictIfNeeded(K key);
	protected abstract StoreEntry<K, V> doPut(K key, StoreEntry<K, V> entry);
	
	//
	protected abstract Map<K, V> doPoll(Filter<K, V> filter);
	protected abstract Map<K, V> doPeek(Filter<K, V> filter);
	
	
	/**
	 * 
	 */
	public AbstractSimpleStore(Cache<K, V> cache, StoreEntry.Factory<K, V> sef) {
		this(cache, sef, 0, null, false);
	}
	
	public AbstractSimpleStore(Cache<K, V> cache, StoreEntry.Factory<K, V> sef, int capacity) {
		this(cache, sef, capacity, null, false);
	}
	
	public AbstractSimpleStore(Cache<K, V> cache, StoreEntry.Factory<K, V> sef, int capacity, ExpirationEvictor.Factory<K, V> ef) {
		this(cache, sef, capacity, ef, false);
	}
	
	public AbstractSimpleStore(Cache<K, V> cache, StoreEntry.Factory<K, V> sef, int capacity, ExpirationEvictor.Factory<K, V> ef, boolean fair) {
		//
		super(cache);
		
		//
		this.entryFactory = sef;
		this.capacity = capacity;
		this.lenient = new AtomicBoolean(true);
		this.lock = new ReentrantReadWriteLock(fair);
		this.evictor = (ef == null ? null : ef.create(this));
	}
	
	/**
	 * 
	 */
	public String toString() {
		return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
		.append("cache", getCache().getName())
		.append("capacity", capacity)
		.append("running", isRunning()).toString();
	}
	
	/**
	 * 
	 */
	protected void doStart() throws Exception {
		//
		if(evictor != null) {
			evictor.start();
		}
	}
	
	protected void doStop() throws Exception {
		//
		if(evictor != null) {
			evictor.stop();
		}
	}
	
	/**
	 * 
	 */
	public final boolean isFair() {
		return lock.isFair();
	}
	
	public final int getCapacity() {
		return capacity;
	}
	
	public final boolean isLenient() {
		return lenient.get();
	}
	
	public final void setLenient(boolean lenient) {
		this.lenient.set(lenient);
	}
	
	public final StoreEntry.Factory<K, V> getEntryFactory() {
		return entryFactory;
	}
	
	public final void lock(boolean exclusive) {
		final Lock l = exclusive ? this.lock.writeLock() : this.lock.readLock();
		l.lock();
	}
	
	public final void unlock(boolean exclusive) {
		final Lock l = exclusive ? this.lock.writeLock() : this.lock.readLock();
		l.unlock();
	}
	
	/**
	 * Store
	 */
	public final int size() {
		lock(false);
		try {
			return doSize();
		} finally {
			unlock(false);
		}
	}
	
	public final boolean containsKey(K key) {
		//
		StoreEntry<K, V> existing = null;
		lock(false);
		try {
			existing = doGet(key);
		} finally {
			unlock(false);
		}
		
		//
		return existing == null || existing.isExpired() ? false : true;
	}
	
	public final void clear() {
		lock(true);
		try {
			doClear();
		} finally {
			unlock(true);
		}
	}
	
	public final V remove(K key) {
		//
		StoreEntry<K, V> existing = null;
		lock(true);
		try {
			existing = doRemove(key);
		} finally {
			unlock(true);
		}
		
		//
		if(existing != null && existing.isExpired()) {
			publish(StoreEventType.expire, existing);
		}
		return existing == null || existing.isExpired() ? null : existing.getValue();
	}
	
	public final V get(K key) {
		//
		StoreEntry<K, V> existing = null;
		lock(false);
		try {
			existing = doGet(key);
		} finally {
			unlock(false);
		}
		
		//
		return existing == null || existing.isExpired() ? null : existing.getValue();
	}
	
	public final Map<K, V> getAll() {
		Map<K, V> r = new HashMap<K, V>();
		lock(false);
		try {
			r = doPeek(new Filter<K, V>() {
				public boolean accept(StoreEntry<K, V> entry) {
					return !entry.isExpired();
				}
			});
		} finally {
			unlock(false);
		}
		return r;
	}

	public final V put(K key, V value) {
		//
		StoreEntry<K, V> evicted = null;
		StoreEntry<K, V> existing = null;
		lock(true);
		try {
			//
			final CacheValidator<K, V> validator = getCache().getValidator();
			if(validator != null) {
				existing = doGet(key);
				V v = existing == null || existing.isExpired() ? null : existing.getValue();
				validator.validate(key, v, value);
			}
			
			//
			evicted = doEvictIfNeeded(key);
			
			//
			existing = doPut(key, entryFactory.create(key, value));
		} finally {
			unlock(true);
		}
		
		//
		if(evicted != null) {
			publish(evicted.isExpired() ? StoreEventType.expire : StoreEventType.evict, evicted);
		}
		if(existing != evicted && existing != null && existing.isExpired()) {
			publish(StoreEventType.expire, existing);
		}
		return existing == null || existing.isExpired() ? null : existing.getValue();
	}
	
	public final V putIfAbsent(K key, V value) {
		//
		StoreEntry<K, V> evicted = null;
		StoreEntry<K, V> existing = null;
		lock(true);
		try {
			existing = doGet(key);
			if(existing == null || existing.isExpired()) {
				//
				final CacheValidator<K, V> validator = getCache().getValidator();
				if(validator != null) {
					validator.validate(key, null, value);
				}
				
				//
				evicted = doEvictIfNeeded(key);

				//
				doPut(key, entryFactory.create(key, value));
			} 
		} finally {
			unlock(true);
		}
		
		//
		if(evicted != null) {
			publish(evicted.isExpired() ? StoreEventType.expire : StoreEventType.evict, evicted);
		}
		if(existing != evicted && existing != null && existing.isExpired()) {
			publish(StoreEventType.expire, existing);
		}
		return existing == null || existing.isExpired() ? null : existing.getValue();
	}
	
	public final Map<K, V> putAll(Map<K, V> values) {
		Map<K, V> r = new HashMap<K, V>(values.size());
		lock(true);
		try {
			for(K key : values.keySet()) {
				try {
					//
					V existing = put(key, values.get(key));
					
					//
					if(existing != null) {
						r.put(key, existing);
					}
				} catch(ValidationException ve) {
					// If validator threw an exception, that exception can be found in "Reject" event
				}
			}
		} finally {
			unlock(true);
		}
		return r;
	}
	
	/**
	 * 
	 */
	public final Map<K, V> detect() {
		//
		Map<K, V> expired = null;
		lock(true);
		try {
			expired = doPoll(new Filter<K, V>() {
				public boolean accept(StoreEntry<K, V> entry) {
					return entry.isExpired();
				}
			});
		} finally {
			unlock(true);
		}
		
		//
		for(Entry<K, V> e : expired.entrySet()) {
			publish(StoreEventType.expire, e.getKey(), e.getValue());
		}
		return expired;
	}
	
	/**
	 * 
	 */
	protected interface Filter<K, V> {
		
		boolean accept(StoreEntry<K, V> entry);
	}
	
	protected abstract static class AbstractFactory<K, V> implements Factory<K, V> {
		//
		protected boolean fair;
		protected StoreEventListener<K, V> listener;
		protected final StoreEntry.Factory<K, V> entryFactory;
		protected final ExpirationEvictor.Factory<K, V> evictorFactory;
		
		/**
		 * 
		 */
		public AbstractFactory(StoreEntry.Factory<K, V> sef) {
			this(sef, null);
		}
		
		public AbstractFactory(StoreEntry.Factory<K, V> sef, ExpirationEvictor.Factory<K, V> ef) {
			this.entryFactory = sef;
			this.evictorFactory = ef;
		}
		
		/**
		 * 
		 */
		public boolean isFair() {
			return fair;
		}

		public void setFair(boolean fair) {
			this.fair = fair;
		}
		
		public StoreEventListener<K, V> getListener() {
			return listener;
		}

		public void setListener(StoreEventListener<K, V> listener) {
			this.listener = listener;
		}
	}
}
